package com.hzmg.akkacloud.worker.starter;

import akka.actor.ActorSystem;
import akka.actor.Address;
import akka.cluster.Cluster;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.hzmg.akkacloud.common.RemoteConstant;
import com.hzmg.akkacloud.common.exception.AkkaCloudException;
import com.hzmg.akkacloud.common.response.ResultDTO;
import com.hzmg.akkacloud.common.serialize.JsonUtils;
import com.hzmg.akkacloud.common.utils.CommonUtils;
import com.hzmg.akkacloud.common.utils.HttpUtils;
import com.hzmg.akkacloud.common.utils.NetUtils;
import com.hzmg.akkacloud.worker.AkkaCloudWorker;
import com.hzmg.akkacloud.worker.common.AkkaCloudWorkerConfig;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * client actorSystem starter
 * @author zbw
 * @since 2022/9/20
 */
public class AkkaStarter {
    private static Logger log= LoggerFactory.getLogger(AkkaStarter.class);
    public static ActorSystem actorSystem;
    private static AkkaCloudWorkerConfig workerConfig;
    public static ActorSystem init(AkkaCloudWorkerConfig config){
        workerConfig=config;
        Stopwatch stopwatch = Stopwatch.createStarted();
        // 初始化 ActorSystem
        Map<String, Object> overrideConfig = Maps.newHashMap();
        overrideConfig.put("akka.remote.artery.canonical.hostname", NetUtils.getLocalHost());
        overrideConfig.put("akka.remote.artery.canonical.port", config.getPort());
        // 设置角色
        Set<String> useRoles = new HashSet<>(config.getUserRoles().isEmpty() ? Collections.singletonList(config.getAppName()) : config.getUserRoles());
        overrideConfig.put("akka.cluster.roles", useRoles);
        Config akkaBasicConfig = ConfigFactory.load(RemoteConstant.WORKER_AKKA_CONFIG_NAME);
        Config akkaFinalConfig = ConfigFactory.parseMap(overrideConfig).withFallback(akkaBasicConfig);

        //为了使用cluster集群，每个worker名字相同
        actorSystem = ActorSystem.create(RemoteConstant.SERVER_ACTOR_SYSTEM_NAME, akkaFinalConfig);
        //创建集群
        Cluster cluster = Cluster.get(actorSystem);
        //加入种子节点 这里直接加入的是服务器节点
        List<Address> nodeList = Lists.newLinkedList();
        config.getServerAddress().forEach(serverAddress -> {
            String[] ipWithPort = serverAddress.split(":");
            nodeList.add(new Address("akka", RemoteConstant.SERVER_ACTOR_SYSTEM_NAME, ipWithPort[0], getPort(serverAddress)));
        });
        log.info("[AKkaCloudWorker] node list:{}", nodeList);
        cluster.joinSeedNodes(nodeList);
        log.info("[AKkaCloudWorker] cluster address:{}", cluster.selfAddress());
        log.info("[Client] Client's akka system started successfully,using time {}.",stopwatch);
        return actorSystem;
    }
    /**
     * Obtain the Port(actorSystem) of the registry。
     */
    private static int getPort(String serverIP) {
        return AkkaCloudWorker.visitWeb(serverIP, workerConfig, log);
    }
}
